package job.gfp.ranks.calc;

import java.io.IOException;
import java.util.Iterator;

import job.gfp.ranks.base.Data;
import job.gfp.ranks.base.RanksUtil;
import mapred.base.TextCombiner;

import org.apache.hadoop.io.Text;

public class Calc_Combiner extends TextCombiner {

    @Override
    protected void combine(Text key, Iterator<Text> values) throws IOException {
        Data sum = new Data();
        while (values.hasNext()) {
            Text value = values.next();
            String text = value.toString().trim();
            if (RanksUtil.isWrappedBy(RanksUtil.Rank, text)) {
                Data data = new Data(RanksUtil.unwrap(RanksUtil.Rank, text));
                sum.add(data);
            } else {
                this.output(key, value);
            }
        }
        this.output(key, new Text(RanksUtil
                .wrap(RanksUtil.Rank, sum.toString())));
    }

    @Override
    public void setUp() {
    }

    @Override
    public void tearDown() throws IOException {
    }

}
